package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.AsyncDimFunction;
import com.atguigu.bean.OrderWide;
import com.atguigu.bean.UserProvinceSku;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

//数据流：web/app -> nginx -> 业务服务器(Mysql) -> Maxwell -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> ClickHouse(DWS)
//程  序： Mock -> Mysql -> Maxwell -> Kafka(ZK) -> OrderDetailApp -> Kafka(ZK) -> UserProvinceSku10sApp(Redis,ZK,HDFS,HBase) -> ClickHouse(ZK)
public class UserProvinceSku10sApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生产环境设置为Kafka分区数

        //        //1.1 开启CheckPoint
        //        env.enableCheckpointing(5 * 60000L);
        //        env.getCheckpointConfig().setCheckpointTimeout(5 * 60000L);
        //        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000L);
        //        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
        //
        //        //1.2 指定状态后端
        //        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/xxxx/xxx"));

        //TODO 2.读取 Kafka dwd_order_wide 主题的数据创建流
        String topic = "dwd_order_wide";
        String groupId = "user_province_sku_10s_app_210927";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer2(topic, groupId));

        //TODO 3.将数据转换为JavaBean对象并过滤为Null的数据(上游是upsert-kafka)
        SingleOutputStreamOperator<OrderWide> orderWideDS = kafkaDS.process(new ProcessFunction<String, OrderWide>() {
            @Override
            public void processElement(String value, Context ctx, Collector<OrderWide> out) throws Exception {
                try {
                    OrderWide orderWide = JSON.parseObject(value, OrderWide.class);
                    out.collect(orderWide);
                } catch (Exception e) {
                    System.out.println("发现脏数据：" + value);
                }
            }
        });

        //TODO 4.使用状态编程的方式，过滤数据(上游是upsert-kafka，不需要活动和购物券相关数据，所以取第一条即可)
        SingleOutputStreamOperator<OrderWide> filterDS = orderWideDS.keyBy(OrderWide::getOrder_detail_id)
                .filter(new RichFilterFunction<OrderWide>() {

                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("value-state", String.class);
                        StateTtlConfig stateTtlConfig = new StateTtlConfig.Builder(Time.seconds(5)) //设置生成环境中最大延迟时间
                                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                                .build();
                        stateDescriptor.enableTimeToLive(stateTtlConfig);
                        valueState = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    public boolean filter(OrderWide value) throws Exception {
                        //取出状态数据
                        String exists = valueState.value();

                        if (exists == null) {
                            //状态为Null,说明当前是该Order_Detail_Id下的第一条到达数据
                            valueState.update("0");
                            return true;
                        } else {
                            return false;
                        }
                    }
                });

        //TODO 5.将数据转换为新的JavaBean对象
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuDS = filterDS.map(orderWide -> {

            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            return UserProvinceSku.builder()
                    .sku_id(orderWide.getSku_id())
                    .sku_name(orderWide.getSku_name())
                    .sku_price(orderWide.getOrder_price())
                    .user_id(orderWide.getUser_id())
                    .province_id(orderWide.getProvince_id())
                    .order_ct(1L)
                    .order_sku_num(orderWide.getSku_num())
                    .order_amount(orderWide.getSplit_total_amount())
                    .ts(sdf.parse(orderWide.getCreate_time()).getTime())
                    .build();
        });

        //TODO 6.提取时间戳生成Watermark,分组开窗聚合
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithWMDS = userProvinceSkuDS.assignTimestampsAndWatermarks(WatermarkStrategy.<UserProvinceSku>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<UserProvinceSku>() {
            @Override
            public long extractTimestamp(UserProvinceSku element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        SingleOutputStreamOperator<UserProvinceSku> reduce10sDS = userProvinceSkuWithWMDS.keyBy(new KeySelector<UserProvinceSku, Tuple3<Long, Long, Long>>() {
            @Override
            public Tuple3<Long, Long, Long> getKey(UserProvinceSku value) throws Exception {
                return new Tuple3<>(value.getSku_id(), value.getUser_id(), value.getProvince_id());
            }
        })
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<UserProvinceSku>() {
                    @Override
                    public UserProvinceSku reduce(UserProvinceSku value1, UserProvinceSku value2) throws Exception {
                        value1.setOrder_ct(value1.getOrder_ct() + value2.getOrder_ct());
                        value1.setOrder_sku_num(value1.getOrder_sku_num() + value2.getOrder_sku_num());
                        value1.setOrder_amount(value1.getOrder_amount().add(value2.getOrder_amount()));
                        return value1;
                    }
                }, new WindowFunction<UserProvinceSku, UserProvinceSku, Tuple3<Long, Long, Long>, TimeWindow>() {
                    @Override
                    public void apply(Tuple3<Long, Long, Long> key, TimeWindow window, Iterable<UserProvinceSku> input, Collector<UserProvinceSku> out) throws Exception {

                        //取出数据
                        UserProvinceSku userProvinceSku = input.iterator().next();

                        //补充窗口信息字段
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        userProvinceSku.setStt(sdf.format(window.getStart()));
                        userProvinceSku.setEdt(sdf.format(window.getEnd()));

                        //输出数据
                        out.collect(userProvinceSku);
                    }
                });

        //打印测试
        reduce10sDS.print("reduce10sDS>>>>>>>>>>>>>");

        //TODO 7.关联维表
//        reduce10sDS.map(userProvinceSku -> {
//            //关联用户维度
//            //关联地区维度
//            //关联商品维度  SKU SPU TM Category1 2 3
//            return userProvinceSku;
//        });

        //7.1 关联用户维表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithUserDS = AsyncDataStream.unorderedWait(reduce10sDS,
                new AsyncDimFunction<UserProvinceSku>("DIM_USER_INFO") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getUser_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) throws Exception {

                        //补充年龄和性别
                        input.setUser_gender(dimInfo.getString("GENDER"));

                        String birthday = dimInfo.getString("BIRTHDAY");
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                        long age = (System.currentTimeMillis() - sdf.parse(birthday).getTime()) / (1000L * 60 * 60 * 24 * 365);

                        input.setUser_age(age);
                    }
                }, 60, TimeUnit.SECONDS);
        //userProvinceSkuWithUserDS.print("userProvinceSkuWithUserDS>>>>>>>>>>>>>>>>");

        //7.2 关联地区维表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithProDS = AsyncDataStream.unorderedWait(userProvinceSkuWithUserDS,
                new AsyncDimFunction<UserProvinceSku>("DIM_BASE_PROVINCE") {
                    @Override
                    public String getKey(UserProvinceSku input) {
                        return input.getProvince_id().toString();
                    }

                    @Override
                    public void join(UserProvinceSku input, JSONObject dimInfo) {
                        //补充省份相关信息
                        input.setProvince_name(dimInfo.getString("NAME"));
                        input.setProvince_area_code(dimInfo.getString("AREA_CODE"));
                        input.setProvince_iso_code(dimInfo.getString("ISO_CODE"));
                        input.setProvince_iso_3166_2(dimInfo.getString("ISO_3166_2"));
                    }
                }, 60, TimeUnit.SECONDS);
        userProvinceSkuWithProDS.print("userProvinceSkuWithProDS>>>>>>>>>>>>>");

        //7.3 关联SKU维表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithSkuDstream = AsyncDataStream.unorderedWait(userProvinceSkuWithProDS, new AsyncDimFunction<UserProvinceSku>("DIM_SKU_INFO") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setSku_name(jsonObject.getString("SKU_NAME"));
                UserProvinceSku.setCategory3_id(jsonObject.getLong("CATEGORY3_ID"));
                UserProvinceSku.setSpu_id(jsonObject.getLong("SPU_ID"));
                UserProvinceSku.setTm_id(jsonObject.getLong("TM_ID"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getSku_id());
            }
        }, 60, TimeUnit.SECONDS);

        //7.4 关联SPU维表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithSpuDstream = AsyncDataStream.unorderedWait(userProvinceSkuWithSkuDstream, new AsyncDimFunction<UserProvinceSku>("DIM_SPU_INFO") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setSpu_name(jsonObject.getString("SPU_NAME"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getSpu_id());
            }
        }, 60, TimeUnit.SECONDS);

        //7.5 关联Category维表 3   2   1
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithCategory3Dstream = AsyncDataStream.unorderedWait(userProvinceSkuWithSpuDstream, new AsyncDimFunction<UserProvinceSku>("DIM_BASE_CATEGORY3") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setCategory3_name(jsonObject.getString("NAME"));
                UserProvinceSku.setCategory2_id(jsonObject.getLong("CATEGORY2_ID"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getCategory3_id());
            }
        }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithCategory2Dstream = AsyncDataStream.unorderedWait(userProvinceSkuWithCategory3Dstream, new AsyncDimFunction<UserProvinceSku>("DIM_BASE_CATEGORY2") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setCategory2_name(jsonObject.getString("NAME"));
                UserProvinceSku.setCategory1_id(jsonObject.getLong("CATEGORY1_ID"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getCategory2_id());
            }
        }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithCategory1Dstream = AsyncDataStream.unorderedWait(userProvinceSkuWithCategory2Dstream, new AsyncDimFunction<UserProvinceSku>("DIM_BASE_CATEGORY1") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setCategory1_name(jsonObject.getString("NAME"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getCategory1_id());
            }
        }, 60, TimeUnit.SECONDS);

        //7.6 关联Trademark维表
        SingleOutputStreamOperator<UserProvinceSku> userProvinceSkuWithTmDstream = AsyncDataStream.unorderedWait(userProvinceSkuWithCategory1Dstream, new AsyncDimFunction<UserProvinceSku>("DIM_BASE_TRADEMARK") {
            @Override
            public void join(UserProvinceSku UserProvinceSku, JSONObject jsonObject) throws Exception {
                UserProvinceSku.setTm_name(jsonObject.getString("TM_NAME"));
            }

            @Override
            public String getKey(UserProvinceSku UserProvinceSku) {
                return String.valueOf(UserProvinceSku.getTm_id());
            }
        }, 60, TimeUnit.SECONDS);
        userProvinceSkuWithTmDstream.print("userProvinceSkuWithTmDstream>>>>>>>>>>>>>>");

        //TODO 8.将数据写出
        userProvinceSkuWithTmDstream.addSink(MyClickHouseUtil.getClickHouseSink("insert into dws_user_province_sku_10s values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        //TODO 9.启动
        env.execute("UserProvinceSku10sApp");

    }
}
